#include <pt.h>
#include "stdio.h"
#include "string.h"
#include "pt_mbus.h"
#include "pt_mbus_cli.h"
#include "osi_api.h"
#include "pt_utils.h"

typedef struct
{
	struct pt_mailbox queue;
	pt_mbus_topic_item_t sItem[PT_TOPIC_MAX];
}pt_mbus_ctx_t;

//上下文信息
static pt_mbus_ctx_t g_s_mbus_ctx;

static int prvFindVaildClientByIndex(int sItemIndex)
{
	pt_mbus_ctx_t* ctx = &g_s_mbus_ctx;

	if(sItemIndex >= PT_ARRAY_SIZE(ctx->sItem)) {
		PT_LOGE("[prvFindVaildClientByIndex]: invalid topic %d", sItemIndex);
		return PT_MBUS_CLIENT_ITEM_COUNT_MAX;
	}

	for (int i = 0; i < PT_MBUS_CLIENT_ITEM_COUNT_MAX; i++)
	{
		pt_mbus_client_item_t* cItem = ctx->sItem[sItemIndex].client[i];
		if (cItem == NULL || cItem->vaild == PT_FALSE)
		{
			return i;
		}
	}

	PT_LOGE("[prvFindVaildClientByIndex]: cannot find empty client %d", sItemIndex);

	return PT_MBUS_CLIENT_ITEM_COUNT_MAX;
}

/**
@brief pt_mbus_init_msg
@param[in,out] msg Message pointer to be initialized
@param[in] body_len Size of message body
@retval Pointer to message body.
*/
uint8_t* pt_mbus_init_msg(pt_mbus_msg_t* msg, uint16_t body_len)
{
	pt_assert(msg);

	msg->len = body_len;

	if (msg->len > SHORT_DATA_MSG_BODY_THRESHOLD)
	{
		int* bodyP = (int*)msg->body;

		*bodyP = (int)pt_malloc(msg->len);
		memset((char*)*bodyP, 0, msg->len);
		return (uint8_t*)*bodyP;
	}
	else
	{
		if (body_len == 0)
		{
			return NULL;
		}
		return msg->body;
	}
}

/**内部函数
@brief free msg body
@param[in,out] msg Message pointer to be initialized
@retval
*/
static void pt_mbus_deinit_msg(pt_mbus_msg_t* msg)
{
	/*长度大于SHORT_DATA_MSG_BODY_THRESHOLD 需要free*/
	if (msg->len > SHORT_DATA_MSG_BODY_THRESHOLD)
	{
		int* bodyP = (int*)msg->body;
		pt_free((char*)*bodyP);
	}
}

/**
@brief Get message body
@param[in] msg Message pointer
@retval Pointer to message body.
*/
uint8_t* pt_mbus_get_msg_body(pt_mbus_msg_t* msg)
{
	pt_assert(msg);

	if (msg->len > SHORT_DATA_MSG_BODY_THRESHOLD)
	{
		int* bodyP = (int*)msg->body;

		return (uint8_t*)*bodyP;
	}
	else
	{
		if (msg->len == 0)
		{
			return NULL;
		}
		return msg->body;
	}
}

/**
@brief pt_mbus_publish
@param[in] topic id
@param[in] len Size of data available
@param[in] data content of data
@retval PT_EOK if successful, otherwise return error number < 0.
*/
int32_t pt_mbus_publish(int topic_id, uint32_t len, void *data)
{
	pt_err_t ret = PT_EOK;
	pt_mbus_ctx_t* ctx = &g_s_mbus_ctx;

	pt_mbus_msg_t *msg = pt_malloc(sizeof(pt_mbus_msg_t));
	uint8_t *buf = pt_mbus_init_msg(msg, len);
	if (buf)
	{
		memcpy(buf, data, len);
	}

	msg->dst_cid = topic_id;

	//将数据发送给client
	ret = pt_mb_send(&ctx->queue, (pt_ubase_t)msg);

	return ret;
}

void pt_mbus_fast_publish(int topic_id, uint32_t len, void *data)
{
	pt_mbus_ctx_t *ctx = &g_s_mbus_ctx;
	pt_mbus_msg_t msg = {
		.dst_cid = topic_id,
	};

	uint8_t *buf = pt_mbus_init_msg(&msg, len);
	if (buf)
	{
		memcpy(buf, data, len);
	}

	pt_mbus_topic_item_t* sItem = &ctx->sItem[msg.dst_cid];

	for (int i = 0; i < PT_MBUS_CLIENT_ITEM_COUNT_MAX; i++)
	{
		pt_mbus_client_item_t* cItem = sItem->client[i];
		if (cItem && cItem->vaild)
		{
						cItem->cbk(pt_mbus_get_msg_body(&msg), cItem->user_data);
		}
	}

	pt_mbus_deinit_msg(&msg);
}

//向server provider添加client信息
int pt_mbus_add_client(int topic_id, pt_mbus_client_item_t* cItem)
{
	pt_mbus_ctx_t* ctx = &g_s_mbus_ctx;

	pt_assert(cItem);
	pt_assert(PT_TOPIC_MAX != topic_id);

	int cItemIndex = prvFindVaildClientByIndex(topic_id);

	if(cItemIndex == PT_MBUS_CLIENT_ITEM_COUNT_MAX) {
		return -1;
	}

	// pt_assert(PT_MBUS_CLIENT_ITEM_COUNT_MAX != cItemIndex);

	ctx->sItem[topic_id].client[cItemIndex] = cItem;

	return cItemIndex;
}

void pt_mbus_remove_client(int topic_id, pt_mbus_client_item_t* cItem)
{
	pt_mbus_ctx_t* ctx = &g_s_mbus_ctx;

	for (int i = 0; i < PT_MBUS_CLIENT_ITEM_COUNT_MAX; i++)
	{
		if (ctx->sItem[topic_id].client[i] == cItem)
		{
			ctx->sItem[topic_id].client[i] = NULL;
		}
	}
}

void pt_mbus_handle(uint32_t timeout)
{
	pt_mbus_ctx_t *ctx = &g_s_mbus_ctx;
	pt_mbus_msg_t *msg;

	pt_ubase_t value;

	uint32_t elapsed;
	osiElapsedTimer_t time;
	osiElapsedTimerStart(&time);

	while (pt_mb_recv(&ctx->queue, &value, timeout) == PT_EOK)
	{
		msg = (pt_mbus_msg_t *)value;
		pt_mbus_topic_item_t* sItem = &ctx->sItem[msg->dst_cid];

		for (int i = 0; i < PT_MBUS_CLIENT_ITEM_COUNT_MAX; i++)
		{
			pt_mbus_client_item_t* cItem = sItem->client[i];
			if (cItem && cItem->vaild)
			{
				cItem->cbk(pt_mbus_get_msg_body(msg), cItem->user_data);
			}
		}

		//释放msg.body的空间
		pt_mbus_deinit_msg(msg);
		pt_free(msg);

		elapsed = osiElapsedTime(&time);
		if (elapsed >= timeout)
		{
			break;
		}

		timeout -= elapsed;
	}
}

void pt_mbus_handle_with_sleep(uint32_t timeout, bool (*is_sys_sleep)(void))
{
	pt_mbus_ctx_t *ctx = &g_s_mbus_ctx;
	pt_mbus_msg_t *msg;

	pt_ubase_t value;

	uint32_t elapsed;
	osiElapsedTimer_t time;
	osiElapsedTimerStart(&time);

	// 睡眠状态就一直等待其他系统事件
	if((*is_sys_sleep)()) timeout = OSI_WAIT_FOREVER;

	while (pt_mb_recv(&ctx->queue, &value, timeout) == PT_EOK)
	{
		msg = (pt_mbus_msg_t *)value;
		pt_mbus_topic_item_t* sItem = &ctx->sItem[msg->dst_cid];

		PT_LOGI("[pt_mbus_handle_with_sleep]: %d %d 0x%02x", (*is_sys_sleep)(), timeout, msg->dst_cid);

		for (int i = 0; i < PT_MBUS_CLIENT_ITEM_COUNT_MAX; i++)
		{
			pt_mbus_client_item_t* cItem = sItem->client[i];
			if (cItem && cItem->vaild)
			{
								cItem->cbk(pt_mbus_get_msg_body(msg), cItem->user_data);
			}
		}

		//释放msg.body的空间
		pt_mbus_deinit_msg(msg);
		pt_free(msg);

		if(timeout == OSI_WAIT_FOREVER) {
			// 如果睡眠状态进入非睡眠状态 就马上退出让ui进行处理
			if(!(*is_sys_sleep)()) break;
		} else {
			elapsed = osiElapsedTime(&time);
			if (elapsed >= timeout)
			{
				break;
			}
			timeout -= elapsed;
		}
	}
}

void pt_mbus_init(void)
{
	pt_mbus_ctx_t* ctx = &g_s_mbus_ctx;

	pt_err_t err = pt_mb_init(&ctx->queue, "dataQu", NULL, 64, PT_IPC_FLAG_FIFO);
	pt_assert(err == PT_EOK);
}
